package com.ljkj.cordial.lengthtcpheartbeat;

import android.os.SystemClock;
import android.text.TextUtils;
import android.util.ArrayMap;
import android.util.Log;

import com.lanjiu.lib.communication.IMSConfig;
import com.lanjiu.lib.communication.util.CheckSumResult;
import com.lanjiu.lib.communication.util.JCRC32;
import com.lanjiu.pro.business.BusinessProtocolMessageStandard;
import com.nic.view.util.JLog;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;

public class TcpLengthHeartbeatClient extends OnEventMessageListener implements OnConnectStateCallback {
    private final static String TAG = TcpLengthHeartbeatClient.class.getSimpleName();

    private static TcpLengthHeartbeatClient myClientH;

    public static TcpLengthHeartbeatClient getInstance() {
        if (myClientH == null) {
            synchronized (TcpLengthHeartbeatClient.class) {
                myClientH = new TcpLengthHeartbeatClient();
            }
        }
        return myClientH;
    }


    private Bootstrap bootstrap;
    private EventLoopGroup eventLoopGroup;
    private volatile String address;
    private volatile int port;
    private volatile boolean forceClose;

    /**
     * 设置重连间隔时间
     */
    private volatile long reconnectSpaceTime = 10;
    /**
     * 重连计数器
     */
    private AtomicInteger reconnectTimes = new AtomicInteger(0);

    private volatile long currentActiveTime;
    private OnDispenseResponseListener onDispenseResponseListener;
    private ArrayMap<String, OnShortResponseListener> observeShorts = new ArrayMap<>();
    private OnHeartbeatReceiveListener onHeartbeatReceiveListener;
    private OnClientConnectedListener onClientConnectedListener;
    private JCRC32 jcrc32;
    private BusinessProtocolMessageStandard.UnifiedEntranceMessage request;
    private boolean isConnected;
    private HashedWheelTimer connectTimer = new HashedWheelTimer();
    private Timeout currentTimeout;

    @Override
    public boolean isForceExit() {
        return forceClose;
    }

    public void setForceClose(boolean forceClose) {
        this.forceClose = forceClose;
    }


    public void addShortResponseListener(String checkSum, OnShortResponseListener listener) {
        removeShortResponseListener(checkSum);
        observeShorts.put(checkSum, listener);
    }

    public void removeShortResponseListener(String checkSum) {
        if (observeShorts.containsKey(checkSum)) {
            observeShorts.remove(checkSum);
        }
    }


    public long getCurrentActiveTime() {
        return currentActiveTime;
    }

    public void setCurrentActiveTime(long currentActiveTime) {
        this.currentActiveTime = currentActiveTime;
    }

    public void setOnHeartbeatReceiveListener(OnHeartbeatReceiveListener onHeartbeatReceiveListener) {
        this.onHeartbeatReceiveListener = onHeartbeatReceiveListener;
    }

    public void setOnClientConnectedListener(OnClientConnectedListener onClientConnectedListener) {
        this.onClientConnectedListener = onClientConnectedListener;
    }

    public void setConnected(boolean connected) {
        isConnected = connected;
    }

    public boolean isConnected() {
        return isConnected;
    }

    public TcpLengthHeartbeatClient() {
        initEventLoop();
    }

    private void initEventLoop() {
        if (eventLoopGroup == null || eventLoopGroup.isShuttingDown() || eventLoopGroup.isShutdown()) {
            eventLoopGroup = new NioEventLoopGroup();
            bootstrap = new Bootstrap();
//        bootstrap.option(ChannelOption.TCP_NODELAY, true);
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new ProtobufVarint32FrameDecoder());
                            pipeline.addLast(new ProtobufDecoder(BusinessProtocolMessageStandard.CheckUnifiedEntranceMessage.getDefaultInstance()));  //要转换的对象实例
                            pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
                            pipeline.addLast(new ProtobufEncoder());
                            pipeline.addLast(new IdleStateHandler(61, 61, 65, TimeUnit.SECONDS)); //netty提供的空闲状态处理器
                            pipeline.addLast(new HeartBeatEventHandler(TcpLengthHeartbeatClient.this, TcpLengthHeartbeatClient.this));
                        }
                    });
        }

    }


    public void connect(String address, int port) {
        this.address = address;
        this.port = port;
        setForceClose(false);
        reconnectTimes.set(0);
        JLog.e(TAG, "connect->doConnect");
        resetConnect(getChannelCxt());
    }

    /**
     * 连接服务端 and 重连
     */
    public void doConnect() {
        if (isForceExit()) {
            close();
            return;
        }
        if (eventLoopGroup.isShutdown()) {
            initEventLoop();
        }

        ChannelFuture connect = null;
        try {
            connect = bootstrap.connect(address, port);
            connect.addListener(new ChannelFutureListener() {

                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    closeTimer();
                    setReConnecting(false);
                    if (channelFuture.isSuccess()) {
                        JLog.d(TAG, "连接成功 ip,port->" + address + "," + port);
                        reconnectSpaceTime = 10;
                        reconnectTimes.set(0);
                        setCurrentActiveTime(SystemClock.elapsedRealtime());
                        setConnected(true);
                        onConnected(channelFuture);
                    } else {
                        if (isForceExit()) {
                            JLog.d(TAG, "断开连接");
                            return;
                        }

                        setConnected(false);
                        if (reconnectTimes.intValue() >= 19) {
                            reconnectSpaceTime = 60;
                        } else if (reconnectTimes.intValue() >= 9) {
                            reconnectSpaceTime = 30;
                        }
                        reconnectTimes.getAndIncrement();
                        JLog.d(TAG, "连接失败[" + reconnectTimes.intValue() + "]次");
                        currentTimeout = connectTimer.newTimeout(new TimerTask() {
                            @Override
                            public void run(Timeout timeout) throws Exception {
                                timeout.cancel();
                                if (!isForceExit() && !isConnected) {
                                    doConnect();
                                }
                            }
                        }, reconnectSpaceTime, TimeUnit.SECONDS);
                    }
                    channelFuture.removeListener(this);
                }
            });
            //如果操作被中断,等待future完成
            connect.awaitUninterruptibly();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //无论future是否成功都会返回CloseFuture，失败的时候会自动关闭Channel
            connect.channel().closeFuture().awaitUninterruptibly();
        }
        return;
    }

    /**
     * 关闭正在重连的定时任务
     */
    public void closeTimer() {
        if (currentTimeout != null) {
            currentTimeout.cancel();
        }
    }

    @Override
    public BusinessProtocolMessageStandard.CheckUnifiedEntranceMessage getHandshakeMsg() {
        BusinessProtocolMessageStandard.CheckUnifiedEntranceMessage u = jcrc32.packageCheckSum(request);
        setServerSendReportCheckSum(u.getChecksum());
        return u;

    }


    @Override
    public void setUserId(String userId) {
        super.setUserId(userId);
    }

    /**
     * 心跳
     */
    public void sendHeartbeat() {
        if (isConnected) {
            send(getEmptyMsg(IMSConfig.HEART_PACKAGE_PHONE));
        }
    }

    /**
     * 断开心跳握手
     */
    public void sendEndHeartbeat() {
        if (isConnected) {
            send(getEmptyMsg(IMSConfig.HEART_PACKAGE_PHONE_END));
        }

    }

    /**
     * 发送不校验的空数据包消息
     */
    public void sendEmptyMsg(String msgType) {
        if (isConnected) {
            send(getEmptyMsg(msgType));
        }

    }

    /**
     * 正常请求
     *
     * @param request
     */
    public void request(BusinessProtocolMessageStandard.UnifiedEntranceMessage request, OnShortResponseListener listener) {
        if (request != null) {
            this.request = request;
            if (isConnected) {
                BusinessProtocolMessageStandard.CheckUnifiedEntranceMessage msg = getHandshakeMsg();
                addShortResponseListener(msg.getUnifiedEntranceMessage().getHead().getMsgType(), listener);
                setServerSendReportCheckSum(msg.getChecksum());
//                send(msg);
                sendBusiness(msg);
            }
        }

    }

    public void setOnDispenseResponseListener(OnDispenseResponseListener onDispenseResponseListener) {
        this.onDispenseResponseListener = onDispenseResponseListener;
    }

    public OnDispenseResponseListener getOnDispenseResponseListener() {
        return this.onDispenseResponseListener;
    }

    @Override
    public void onConnected(ChannelFuture ctx) {
        jcrc32 = new JCRC32();
        if (onClientConnectedListener != null) {
            onClientConnectedListener.onConnected();
        }
    }


    @Override
    public void onDoConnect(ChannelHandlerContext ctx) {
        if (isForceExit()) {
            return;
        }
        JLog.e(TAG, "write->doConnect");
        resetConnect(ctx);
    }

    private void resetConnect(ChannelHandlerContext ctx) {
        try {
            if (isReConnecting()) {
                JLog.e(TAG, "-->doConnecting");
                return;
            }
            setConnected(false);
            setReConnecting(true);
            reconnectTimes.set(0);
            if (ctx != null && !ctx.executor().isShutdown() && ctx.channel().isActive()) {
                ctx.close().awaitUninterruptibly();
                JLog.e(TAG, "-->doConnect-close");
            } else {
                JLog.e(TAG, "-->doConnect-no channel");
            }
            doConnect();
        } catch (Exception e) {
        }
    }

    public void close() {
        try {
            setConnected(false);
            if (getChannelCxt() != null) {
                getChannelCxt().close();
            }
            if (eventLoopGroup != null) {
                eventLoopGroup.shutdownGracefully();
            }
        } catch (Exception e) {
            throw new RuntimeException("-shutdown-");
        }
    }

    /**
     * 主动重连
     */
    public void onReConnect() {
        JLog.e(TAG, "i reset->doConnect");
        setForceClose(false);
        resetConnect(getChannelCxt());
    }

    @Override
    public void close(ChannelHandlerContext ctx) {
        JLog.e(TAG, "close");
        ctx.close();
        close();
    }

    @Override
    public void onRead(BusinessProtocolMessageStandard.CheckUnifiedEntranceMessage msg) {
        BusinessProtocolMessageStandard.UnifiedEntranceMessage uf = msg.getUnifiedEntranceMessage();
        BusinessProtocolMessageStandard.Head head = uf.getHead();
        CheckSumResult checkResult = jcrc32.checkSumValidateProtocol(msg, uf);
        if (checkResult.b_result && !TextUtils.isEmpty(head.getFromId())) {
//            if (head.getFromId().length() == 9) {
//            } else {
//            }
            if (onDispenseResponseListener != null) {
                onDispenseResponseListener.onResponse(head.getMsgType(), msg);
            }
            if (observeShorts.containsKey(head.getMsgType())) {
                observeShorts.get(head.getMsgType()).onResponse(head.getMsgType(), msg);
                removeShortResponseListener(head.getMsgType());
            }

        } else {
//            onFailed(head.getMsgType());
            if (!TextUtils.isEmpty(head.getFromId())) {
//                if (head.getFromId().length() == 9) {
//                } else {
//                }
                if (onDispenseResponseListener != null) {
                    onDispenseResponseListener.onFailed(head.getMsgType());
                }
                if (observeShorts.containsKey(head.getMsgType())) {
                    observeShorts.get(head.getMsgType()).onFailed(head.getMsgType());
                    removeShortResponseListener(head.getMsgType());
                }
            }

        }

    }

    @Override
    public void onFailed(String msgType) {
        Log.d(TAG, "receive data fail->" + msgType);
    }

    @Override
    public void receiveHeartbeat(String fromId) {
        if (onHeartbeatReceiveListener != null) {
            onHeartbeatReceiveListener.receiveHeartbeat(fromId);
        }
    }

}
